package com.atguigu.Flink.state;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.POJO.UrlViewCount;
import com.atguigu.Flink.POJO.WordCount;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;

public class Flink14_TopN_KeyedProcessFunction {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.getTs();
                            }
                        }));
        ds.print("input");
        //统计10秒内的url的top2
        //通过窗口统计每个url的点击次数（多并行的实现）
        SingleOutputStreamOperator<UrlViewCount> urlViewCountDs = ds.map(
                event -> new WordCount(event.getUrl(), 1L)
        ).keyBy(
                k -> k.getWord()
        ).window(
                TumblingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))
        ).aggregate(
                new AggregateFunction<WordCount, Long, UrlViewCount>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(WordCount wordCount, Long aLong) {
                        return aLong + 1;
                    }

                    @Override
                    public UrlViewCount getResult(Long aLong) {
                        return new UrlViewCount(null, aLong, null, null);
                    }

                    @Override
                    public Long merge(Long aLong, Long acc1) {
                        return null;
                    }
                }
                ,
                new ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context, Iterable<UrlViewCount> iterable, Collector<UrlViewCount> collector) throws Exception {
                        //迭代数据
                        UrlViewCount urlViewCount = iterable.iterator().next();
                        //补充信息
                        urlViewCount.setUrl(s);
                        urlViewCount.setStart(context.window().getStart());
                        urlViewCount.setEnd(context.window().getEnd());
                        collector.collect(urlViewCount);
                    }
                }
        );

        urlViewCountDs.print();

        //将窗口计算的结果收集到一起，排序取topN
        urlViewCountDs.keyBy(
                k->k.getEnd()
        ).process(
                new KeyedProcessFunction<Long, UrlViewCount, String>() {
                    //声明状态，存储数据
                    private ListState<UrlViewCount> listState;

                    //声明状态，标识定时器是否已经注册
                    private ValueState<Boolean> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //初始化状态
                        ListStateDescriptor<UrlViewCount> url = new ListStateDescriptor<>("url", UrlViewCount.class);
                        listState = getRuntimeContext().getListState(url);
                        ValueStateDescriptor<Boolean> valueStateDescriptor = new ValueStateDescriptor<>("valueState", Boolean.class);
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(UrlViewCount urlViewCount, KeyedProcessFunction<Long, UrlViewCount, String>.Context context, Collector<String> collector) throws Exception {
                        //将来的数据存到状态
                        listState.add(urlViewCount);
                        //注册定时器
                        //判断定时器是否已经注册
                        if (valueState.value() == null || valueState.value() == false){
                            context.timerService().registerEventTimeTimer(urlViewCount.getEnd());
                            //更新状态
                            valueState.update(true);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        //定时器触发，表示该收集的数据都已经收集齐了，排序取topN
                        //从状态中取出数据
                        Iterable<UrlViewCount> iterable = listState.get();
                        ArrayList<UrlViewCount> viewCountArrayList = new ArrayList<>();
                        for (UrlViewCount urlViewCount : iterable) {
                            viewCountArrayList.add(urlViewCount);
                        }
                        //排序
                        viewCountArrayList.sort(
                                new Comparator<UrlViewCount>() {
                                    @Override
                                    public int compare(UrlViewCount o1, UrlViewCount o2) {
                                        return Long.compare(o2.getCount(),o1.getCount());
                                    }
                                }
                        );
                        //取top2
                        StringBuilder resultBuilder = new StringBuilder("==================Top2==================\n");
                        for (int i = 0; i < Math.min(2,viewCountArrayList.size()); i++) {
                            UrlViewCount urlViewCount = viewCountArrayList.get(i);
                            resultBuilder.append("NO." + (i+1)  + " URL: " + urlViewCount.getUrl() + " , 点击次数: " + urlViewCount.getCount() + "\n") ;
                        }
                        resultBuilder.append("========================================\n") ;
                        //输出结果
                        out.collect(resultBuilder.toString());
                    }
                }
        ).print("top2");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
